package com.jy.rhin.domain.notification.publish;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONPath;
import com.jy.framework.base.utils.AssertUtils;
import com.jy.framework.base.utils.Converter;
import com.jy.framework.base.utils.DateUtil;
import com.jy.rhin.application.infra.event.MessageSaveEvent;
import com.jy.rhin.domain.exception.RhinInfraErrorType;
import com.jy.rhin.domain.notification.publish.entity.PublishMessageDO;
import com.jy.rhin.domain.notification.publish.entity.PublishMessageStatisticsDO;
import com.jy.rhin.domain.notification.publish.entity.SubscribeAndFilterVO;
import com.jy.rhin.infra.repo.notfication.publish.query.PublishMessageQueryHandler;
import com.jy.rhin.infra.service.CacheService;
import com.jy.rhin.infra.uitl.Asserts;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author WeiQ
 * @date 2022/6/29 8:53
 * 派发的消息记录域服务
 */
@Slf4j
@Service
public class PublishMessageDomainService {

    /**
     * 如果是无过滤条件订阅主题，则SubscribeAndFilterVO.getQueryFilterExp为null，防止分组函数NPE，则设置一个特殊的key。
     */
    private static final String FILTER_EMPTY_KEY = "NULL";
    @Resource
    PublishMessageDao publishMessageDao;
    @Resource
    CacheService cacheService;
    @Resource
    PublishMessageQueryHandler publishMessageQueryHandler;
    @Resource
    ApplicationEventPublisher applicationEventPublisher;

    private static final String DATE_FORMAT = "yyyy-MM-dd";

    /**
     * 发布消息给订阅者：在收到某个主题的消息后，找到订阅该主题的所有订阅者，为每一个订阅者创建一条发布消息记录
     *
     * @param subscribeAndFilterList
     * @param messageId
     * @param message
     * @return
     */
    public List<PublishMessageDO> buildPublishMessage(List<SubscribeAndFilterVO> subscribeAndFilterList, Long messageId, String message) {
        //根据订阅者分组，分别处理每个订阅者的过滤条件，是否需要发送消息
        //针对订阅者处理，根据过滤条件分组，必须满足所有过滤条件才发送消息；如果一个过滤条件中包含多个值，只需要满足其中一个
        Map<String, Map<String, List<SubscribeAndFilterVO>>> subscribeMapBySubscriberCode = subscribeAndFilterList.stream().collect(Collectors.groupingBy(SubscribeAndFilterVO::getSubscriberCode, HashMap::new,
                //如果过滤条件的分组map包含NULL的key，说明是无条件的订阅
                Collectors.groupingBy(vo -> StringUtils.defaultIfBlank(vo.getQueryFilterExp(), FILTER_EMPTY_KEY))));
        //根据订阅者code，将订阅记录转换成map，订阅记录中订阅者/发布者/主题相关字段是相同的
        Map<String, SubscribeAndFilterVO> subscriberMap = subscribeAndFilterList.stream().collect(Collectors.toMap(SubscribeAndFilterVO::getSubscriberCode, vo -> vo, (t, t2) -> t));
        //创建需要通知给订阅者的记录
        List<PublishMessageDO> records = subscribeMapBySubscriberCode.entrySet().stream().filter(subscribeEntry -> compareFilter(message, subscribeEntry.getValue()))
                //根据过滤条件，筛选出所有满足预期条件的订阅者
                .map(Map.Entry::getKey)
                //获取订阅者/发布者/主题等信息
                .map(subscriberMap::get)
                //为筛选后的订阅者创建发布消息记录
                .map(vo -> new PublishMessageDO(vo).init(messageId)).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(records)) {
            return Collections.emptyList();
        }
        //批量保存
        publishMessageDao.save(records);
        return records;
    }

    /**
     * 比较订阅时设置的过滤条件，是否满足预期：需要满足所有过滤条件，每一个过滤条件可能包含1项或多项
     * 如以下过滤条件: (a = 1 or a = 2) and b = 3 ，共两个过滤条件，其中第一个条件包含两项；第二个条件包含一项
     *
     * @param message
     * @param subscribeMapByFilterExp
     * @return
     */
    private boolean compareFilter(String message, Map<String, List<SubscribeAndFilterVO>> subscribeMapByFilterExp) {
        //订阅者订阅主题，只要有过滤条件，则map一定不包含NULL的key；
        //如果过滤条件为空，说明接收该主题所有数据，直接返回true
        if (subscribeMapByFilterExp.containsKey(FILTER_EMPTY_KEY)) {
            return true;
        }
        for (List<SubscribeAndFilterVO> subscribeAndFilterListOfFilterExp : subscribeMapByFilterExp.values()) {
            boolean compareFilterExp = compareFilterExp(message, subscribeAndFilterListOfFilterExp);
            if (!compareFilterExp) {
                return false;
            }
        }
        return true;
    }

    /**
     * 比较订阅时设置的过滤条件，是否满足预期:只需要满足其中一项即可
     *
     * @param message
     * @param subscribeListByFilterExp 相同过滤条件的集合 用OR连接
     * @return
     */
    private boolean compareFilterExp(String message, List<SubscribeAndFilterVO> subscribeListByFilterExp) {
        //订阅主题，如果有过滤条件，则至少设置一个目标值。如果没有设置值，标识订阅流程异常。
        // 正常情况下有过滤条件没有目标值的订阅，会抛异常
        if (CollectionUtils.isEmpty(subscribeListByFilterExp)) {
            return false;
        }
        //获取该过滤条件的jsonPath，从message获取当前消息的值
        String filterJsonPath = subscribeListByFilterExp.get(0).getFilterJsonPath();
        Object object = JSONPath.read(message, filterJsonPath, Object.class);
        if (Objects.isNull(object)) {
            return false;
        }
        if (object instanceof JSONArray) {
            JSONArray array = JSON.parseArray(JSON.toJSONString(object));
            for (Object obj : array) {
                if (compareExp(subscribeListByFilterExp, obj)) {
                    return true;
                }
            }
            return false;
        }
        return compareExp(subscribeListByFilterExp, object);
    }

    private boolean compareExp(List<SubscribeAndFilterVO> subscribeListByFilterExp, Object object) {
        String value = object.toString();
        for (SubscribeAndFilterVO vo : subscribeListByFilterExp) {
            //比较message中的value与订阅时设置的期望值
            boolean compare = vo.getCompareExp().compare(value, vo.getQueryFilterValue());
            if (compare) {
                //只要满足其中一个值，则该过滤条件满足
                return true;
            }
        }
        return false;
    }

    /**
     * 发布消息，记录发送操作结果
     *
     * @param publishId
     * @param publishResult
     * @param failReason
     */
    @Transactional(rollbackFor = Exception.class)
    public void publishMessage(Long publishId, boolean publishResult, String failReason) {
        PublishMessageDO publishMessage = publishMessageDao.queryById(publishId);
        Asserts.nonNull(publishMessage, RhinInfraErrorType.PUBLISH_MESSAGE_NOT_FOUND);
        boolean updateSuccess = publishResult ? publishMessage.publishSuccess() : publishMessage.publishFail(failReason);
        if (updateSuccess) {
            publishMessageDao.save(publishMessage);
            applicationEventPublisher.publishEvent(new MessageSaveEvent());
        }
    }

    /**
     * 重置对应时间段的消息发送统计
     *
     * @param startDateStr
     * @param endDateStr
     */
    @Transactional(rollbackFor = Exception.class)
    public void resetStatistics(String startDateStr, String endDateStr) {
        AssertUtils.notTrue(StringUtils.isEmpty(startDateStr) || StringUtils.isEmpty(endDateStr), RhinInfraErrorType.START_END_DATE_NOT_NULL);
        Date startDate = Converter.toDate(startDateStr, DATE_FORMAT);
        Date endDate = Converter.toDate(endDateStr, DATE_FORMAT);
        publishMessageDao.resetStatistics(startDate, endDate);
    }

    /**
     * 消息通知数缓存+1
     */
    public void addNowDateTotalCache() {
        String nowDate = Converter.toDateStr(new Date());
        cacheService.addNowDateTotalCache(nowDate);
    }

    /**
     * 删除当天的消息通知数缓存
     */
    public void removeNowDateTotalCache() {
        String nowDate = Converter.toDateStr(new Date());
        cacheService.removeNowDateTotalCache(nowDate);
    }

    /**
     * 设置当天的消息通知数缓存
     */
    public void setNowDateTotalCache() {
        String nowDate = Converter.toDateStr(new Date());
        Integer total = publishMessageQueryHandler.queryNowDateTotal(nowDate);
        cacheService.setNowDateTotalCache(nowDate, total);
    }

    /**
     * 获取当天的消息通知数缓存
     *
     * @return
     */
    public Integer getNowDateTotalCache() {
        String nowDate = Converter.toDateStr(new Date());
        return cacheService.getNowDateTotalCache(nowDate);
    }

    /**
     * 保存昨天的消息通知数量
     */
    public void saveYesterdayTotal() {
        Date yesterday = DateUtil.dateAddDays(new Date(), -1);
        String yesterdayStr = Converter.toDateStr(yesterday);
        Integer total = publishMessageQueryHandler.queryNowDateTotal(yesterdayStr);
        if (Objects.nonNull(total) && total != 0) {
            Date yesterdayFormat = Converter.toDate(yesterdayStr, DATE_FORMAT);
            publishMessageDao.saveStatistics(new PublishMessageStatisticsDO(yesterdayFormat, total));
        }
    }

}
